package com.lagou.rpc.consumer.client;


import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.lagou.rpc.consumer.bean.ServiceMeta;
import com.lagou.rpc.consumer.bean.ServiceProvider;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ThreadLocalRandom;
import java.util.stream.Collectors;

@Slf4j
@Component
public class RpcClientHolder {

    private static final String SERVICES_PATH = "/lg-rpc/services";
    
    @Autowired
    private CuratorFramework curatorFramework;

    private List<RpcClient> rpcClients = new ArrayList<>();

    /**
     * 服务元数据，记录请求耗时等等
     */
    private Map<String, ServiceMeta> serviceMetaMap = Maps.newHashMap();

    @PostConstruct
    public void init() {
        try {
            refreshRpcClients();
            PathChildrenCache cache = new PathChildrenCache(curatorFramework, SERVICES_PATH, true);
            cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
            cache.getListenable().addListener((client, event) -> {
                switch (event.getType()) {
                    case CHILD_ADDED:
                    case CHILD_REMOVED:
                        refreshRpcClients();
                        break;
                    case CHILD_UPDATED:
                        refreshServiceMeta(event.getData());
                        break;
                    default:
                        break;
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void refreshServiceMeta(ChildData childData) {
        String path = childData.getPath();
        String[] arr = path.split("/");
        String childPath = arr[arr.length-1];
        String data = new String(childData.getData());
        serviceMetaMap.put(childPath, JSON.parseObject(data, ServiceMeta.class));
        log.info("refreshServiceMeta, path:{}, data:{}", childPath, data);
    }

    private void refreshRpcClients() throws Exception {
        List<String> services = curatorFramework.getChildren().forPath(SERVICES_PATH);
        List<ServiceProvider> newProviders = services.stream().map(s -> {
            String[] ss = s.split(":");
            return new ServiceProvider(ss[0], Integer.valueOf(ss[1]));
        }).collect(Collectors.toList());

        // 移除掉下线的服务
        List<ServiceProvider> aliveProviders = Lists.newArrayList();
        Iterator<RpcClient> it = rpcClients.iterator();
        while (it.hasNext()) {
            RpcClient next = it.next();
            if (!newProviders.contains(next.getServiceProvider())) {
                next.close(); // 关闭客户端连接，避免连接泄露
                it.remove();
                log.info("remove offline provider: {}", next.getServiceProvider());
            } else {
                aliveProviders.add(next.getServiceProvider());
            }
        }

        // 添加新增的服务
        newProviders.stream()
                .filter(p -> !aliveProviders.contains(p))
                .forEach(p -> {
                    rpcClients.add(new RpcClient(p));
                    log.info("register alive provider: {}", p);
                });
    }

    public void updateServiceLastCallMeta(RpcClient rpcClient, long durationMs) throws Exception {
        ServiceMeta serviceMeta = new ServiceMeta(durationMs, System.currentTimeMillis());
        String path = SERVICES_PATH + "/" + rpcClient.getServiceProvider().print();
        curatorFramework.setData().forPath(path, JSON.toJSONString(serviceMeta).getBytes());
    }

    public RpcClient client() {
        long minCallDuration = 0L;
        RpcClient selectedClient = null;
        for (RpcClient rpcClient : rpcClients) {
            ServiceMeta serviceMeta = serviceMetaMap.get(rpcClient.getServiceProvider().print());
            // 优先未请求过的
            if (serviceMeta == null || serviceMeta.getLastCallTimestamp() == 0) {
                selectedClient = rpcClient;
                break;
            }
            // 距离上一次请求时间大于5秒
            if (System.currentTimeMillis() - serviceMeta.getLastCallTimestamp() > 60 * 1000L) {
                selectedClient = rpcClient;
                break;
            }
            // 取上一次请求时间最短的
            if (minCallDuration == 0 || serviceMeta.getLastCallDurationMs() < minCallDuration) {
                minCallDuration = serviceMeta.getLastCallDurationMs();
                selectedClient = rpcClient;
            }
        }
        if (selectedClient != null) {
            return selectedClient;
        }
        // 默认随机取一个
        return rpcClients.get(ThreadLocalRandom.current().nextInt(rpcClients.size()));
    }

}
